package demo.base.connect;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.util.Collector;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;

public class OrderDetail extends Order {
    private String product;
    private double sum;

    public OrderDetail(String orderId, String product, double sum) {
        super(orderId);
        this.product = product;
        this.sum = sum;
    }

    public static OrderDetail of(String orderId, String product, double sum) {
        return new OrderDetail(orderId, product, sum);
    }

    public static OrderDetail empty() {
        return new OrderDetail("", "", 0);
    }

    @Override
    public String toString() {
        return JSON.toJSONString(this, SerializerFeature.WriteMapNullValue);
    }

    public double getSum() {
        return sum;
    }

    public void setSum(double sum) {
        this.sum = sum;
    }

    public String getProduct() {
        return product;
    }

    public void setProduct(String product) {
        this.product = product;
    }

    public static OrderDetailMapFunction getMapFunction() {
        return new OrderDetailMapFunction();
    }

    private static class OrderDetailMapFunction extends RichMapFunction<String, OrderDetail> {
        @Override
        public OrderDetail map(String value) throws Exception {
            String[] splits = value.split(",");
            if (splits.length != 3) {
                System.out.println("OrderDetailMapFunction input error:" + value);
                return OrderDetail.empty();
            } else {
                double sum;
                try {
                    sum = Double.parseDouble(splits[2]);
                } catch (NumberFormatException E) {
                    E.printStackTrace();
                    System.out.println("NumberFormatException: " + E.getMessage());
                    return OrderDetail.empty();
                }
                return OrderDetail.of(splits[0], splits[1], sum);
            }
        }
    }

    public static OrderDetailFlatMapFunction getFlatMapFunction() {
        return new OrderDetailFlatMapFunction();
    }

    private static class OrderDetailFlatMapFunction extends RichFlatMapFunction<String, String> {
        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            String[] splits = value.split(",");
            if (splits.length != 3) {
                System.out.println("OrderDetailMapFunction input error:" + value);
            } else {
                JSONObject jsonObject = new JSONObject();
                String orderId = splits[0];
                String product = splits[1];
                String sum = splits[2];

                jsonObject.put("orderId", orderId);
                jsonObject.put("product", product);
                out.collect(jsonObject.toJSONString());

                jsonObject.clear();

                jsonObject.put("orderId", orderId);
                jsonObject.put("sum", sum);
                out.collect(jsonObject.toJSONString());
            }
        }
    }
}
